#pragma once
#include <condition_variable>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>

#include "Common/UUIDHelper.hpp"
#include "Client_consumer.hpp"
#include "Message.pb.h"
#include "Subscriber.hpp"
#include "codec/codec.h"
#include "fmtlog/fmtlog.h"
#include "google/protobuf/message.h"
#include "mq_proto.pb.h"
#include "muduo/net/Callbacks.h"

namespace MY_MQ {
using MessagePtr              = std::shared_ptr<google::protobuf::Message>;
using ProtobufCodecPtr        = std::shared_ptr<ProtobufCodec>;
using basicConsumeResponsePtr = std::shared_ptr<MQ_Protobuf::basicConsumeResponse>;
using basicCommonResponsePtr  = std::shared_ptr<MQ_Protobuf::basicCommonResponse>;

class ClientChannel {
public:
    using self_ptr = std::shared_ptr<ClientChannel>;

    ClientChannel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
        : _cid(UUIDHelper::uuidWithShuffledMAC()), _conn(conn), _codec(codec) {
    }

    bool OpenChannel() {
        std::string                     rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::openChannelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }

    void CloseChannel() {
        std::string                      rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::closeChannelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        waitResponse(rid);
        return;
    }

    bool declareExchange(const std::string& name, MQ_Protobuf::ExchangeType type, bool durable, bool auto_delete,
                         google::protobuf::Map<std::string, std::string>& args) {
        std::string                         rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::declareExchangeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(name);
        req.set_exchange_type(type);
        req.set_durable(durable);
        req.set_auto_delete(auto_delete);
        req.mutable_args()->swap(args);

        _codec->send(_conn, req);
        // 这里需要等待响应
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }

    void deleteExchange(const std::string& name) {
        std::string                        rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::deleteExchangeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(name);
        _codec->send(_conn, req);
        waitResponse(rid);
    }

    bool DeclareQueue(const std::string QueueName, bool QueueDurable, bool QueueExclusive, bool QueueAutoDelete,
                      google::protobuf::Map<std::string, std::string>& QueueArgs) {
        std::string                      rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::declareQueueRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(QueueName);
        req.set_durable(QueueDurable);
        req.set_exclusive(QueueExclusive);
        req.set_auto_delete(QueueAutoDelete);
        req.mutable_args()->swap(QueueArgs);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }

    void DeleteQueue(const std::string& QueueName) {
        std::string                     rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::deleteQueueRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(QueueName);
        _codec->send(_conn, req);
        waitResponse(rid);
    }

    bool QueueBind(const std::string& ExchangeName, const std::string& QueueName, const std::string& Key) {

        std::string                   rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::queueBindRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ExchangeName);
        req.set_queue_name(QueueName);
        req.set_binding_key(Key);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }

    void QueueUnbind(const std::string& ExchangeName, const std::string& QueueName) {
        std::string                     rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::queueUnBindRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ExchangeName);
        req.set_queue_name(QueueName);
        _codec->send(_conn, req);
        waitResponse(rid);
    }

    void basicPublish(const std::string& ExhcangeName, const MQ_Protobuf::BasicProperties* bp,
                      const std::string& body) {
        std::string                      rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::basicPublishRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ExhcangeName);
        req.set_body(body);

        if (bp != nullptr) {
            req.mutable_properties()->set_id(bp->id());
            req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            req.mutable_properties()->set_routing_key(bp->routing_key());
        }
        _codec->send(_conn, req);
        waitResponse(rid);
    }

    void basicAck(const std::string& MessageId) {
        if (nullptr == _consumer) {
            logd("consumer is null when dealing the message");
            return;
        }
        std::string                  rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::basicAckRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(_consumer->qname);
        req.set_message_id(MessageId);
        _codec->send(_conn, req);
        waitResponse(rid);
    }

    void BasicCancel() {
        if (nullptr == _consumer) {
            logd("consumer is null when dealing the message");
            return;
        }
        std::string                     rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::basicCancelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_consumer_tag(_consumer->tag);
        _codec->send(_conn, req);
        waitResponse(rid);
    }

    bool basicConsume(const std::string& consumerID, const std::string QueueName, bool AutoAck,
                      const ConsumerCallback& cb) {
        if (nullptr != _consumer) {
            logd("当前信道已经订阅其他队列消息");
            return false;
        }
        std::string                      rid = UUIDHelper::uuidWithShuffledMAC();
        MQ_Protobuf::basicConsumeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(QueueName);
        req.set_consumer_tag(consumerID);
        req.set_auto_ack(AutoAck);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        if (resp->ok() == false) {
            logd("添加多订阅消息失败");
            return false;
        }
        _consumer = std::make_shared<Consumer>(consumerID, QueueName, AutoAck, cb);
        return true;
    }

    // 向_basic_common_response_map中添加resp，并通知所有等待的线程
    void putBasicReponse(const basicCommonResponsePtr& resp) {
        // 创建一个std::unique_lock对象，用于锁定互斥量
        std::unique_lock<std::mutex> lock(_mtx);

        // 将resp添加到_basic_common_response_map中
        _basic_common_response_map[resp->rid()] = resp;
        // 通知所有等待的线程
        _cv.notify_all();
    }

    void consume(const basicConsumeResponsePtr& resp) {
        if (nullptr == _consumer) {
            logd("consumer is null when dealing the message");
            return;
        }
        if (resp->consumer_tag() != _consumer->tag) {
            logd("consumer_tag is not match");
            return;
        }
        _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
    }

    std::string cid() const {
        return _cid;
    }

private:
    basicCommonResponsePtr waitResponse(const std::string& rid) {
        std::unique_lock<std::mutex> lock(_mtx);
        _cv.wait(lock, [rid,this]() {
            return _basic_common_response_map.find(rid) != _basic_common_response_map.end();
        });
        auto resp = _basic_common_response_map[rid];
        _basic_common_response_map.erase(rid);
        return resp;
    }

private:
    std::string                                                  _cid;
    muduo::net::TcpConnectionPtr                                 _conn;
    ProtobufCodecPtr                                             _codec;
    Consumer::self_ptr                                           _consumer;
    std::mutex                                                   _mtx;
    std::condition_variable                                      _cv;
    std::pmr::unordered_map<std::string, basicCommonResponsePtr> _basic_common_response_map;
};

class ClientChannelManager {
public:
    using self_ptr = std::shared_ptr<ClientChannelManager>;

    ClientChannel::self_ptr create(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec) {
        std::unique_lock<std::mutex> lock(_mtx);
        auto                         channel = std::make_shared<ClientChannel>(conn, codec);
        _channles[channel->cid()]            = channel;
        return channel;
    }

    void remoeve(const std::string& cid) {
        std::unique_lock<std::mutex> lock(_mtx);
        _channles.erase(cid);
    }

    ClientChannel::self_ptr get(const std::string& cid) {
        std::unique_lock<std::mutex> lock(_mtx);
        if (_channles.find(cid) != _channles.end()) {
            return _channles[cid];
        }
        return nullptr;
    }

private:
    std::mutex                                               _mtx;
    std::unordered_map<std::string, ClientChannel::self_ptr> _channles;
};
};  // namespace MY_MQ